﻿using Confluent.Kafka;
using log4net.Appender;
using log4net.Core;
using System;
using System.Collections.Generic;
using System.Net;
using System.Linq;

namespace log4net.extension.kafka
{
    public class KafkaAppender : AppenderSkeleton
    {
        private Producer<Null, Log4netLogEntity> m_producer = null;
        private string m_localIp = null;
        public override void ActivateOptions()
        {
            base.ActivateOptions();
            var config = new Dictionary<string, object>
                {
                    { "bootstrap.servers", BrokerList }
                };
            m_producer = new Producer<Null, Log4netLogEntity>(config, null, new LogEntitySerializer());
            m_localIp = GetLocalIp();
        }

        protected override void Append(Core.LoggingEvent loggingEvent)
        {
            try
            {
                if (m_producer == null) return;
                var message = GetLogEntity(loggingEvent);
                m_producer.ProduceAsync(this.TopicName, null, message);
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("Exception while writing to kafka", ex);
            }
        }

        private Log4netLogEntity GetLogEntity(LoggingEvent loggingEvent)
        {
            string message = string.Empty;
            string exceptionStr = loggingEvent.GetExceptionString();
            if (exceptionStr != null && exceptionStr.Length > 0)
                message = string.Format("{0}\r\n{1}", loggingEvent.RenderedMessage, exceptionStr);
            else
                message = loggingEvent.RenderedMessage;

            return new Log4netLogEntity
            {
                LogTime = loggingEvent.TimeStamp.ToString("yyyy-MM-dd HH:mm:ss"),
                Level = loggingEvent.Level.DisplayName,
                ServerIp = m_localIp,
                Message = message
            };
        }
        private string GetLocalIp()
        {
            var ipArray = Dns.GetHostEntry(Dns.GetHostName()).AddressList;
            if (ipArray != null && ipArray.Length > 0)
            {
                var ip4 = ipArray.FirstOrDefault(p => p.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork);
                if (ip4 != null)
                {
                    return ip4.ToString();
                }
            }
            return string.Empty;
        }

        protected override void OnClose()
        {
            base.OnClose();
            if (m_producer != null)
            {
                m_producer.Flush(TimeSpan.FromSeconds(3));
                m_producer.Dispose();
                m_producer = null;
            }
        }

        #region Property

        public string TopicName { get; set; }
        public string BrokerList { get; set; }

        #endregion
    }

}
